package com.shujia.flink.core
import  org.apache.flink.streaming.api.scala._
import  org.apache.flink.api.common.RuntimeExecutionMode
object Demo02BatchWC {
  def main(args: Array[String]): Unit = {
    //创建Flink环境

    //获取一个环境，flink进行自动判断是本地还是集群
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    /**
     * 修改flink处理模式
     * 流处理：一条一条处理，每次处理会输出一个结果
     * 批处理：一起处理，输出最终结果，批处理模式不能用于处理无界流的数据
     *
     *
     * 有界流：比如读取文件，有开始，有结束
     * 无界流：比如socket,有开始无结束
     *
     *
     *
     *
     * flink流批统一：同一套代码既能做流处理，也能做批处理
     *
     *
     * 当使用批处理模式时，flink底层处理数据的方式和spark一样，先 执行上游task再执行下游task
     *
     *
     *
     */

    //默认是流处理
    env.setRuntimeMode(RuntimeExecutionMode.BATCH)


    /**
     * 使用flink处理文件
     *
     */



    //读取文件  有界流
    val linesDS: DataStream[String] = env.readTextFile("data/words.txt")

    //1. 将一行转换成多行
    val wordsDS: DataStream[String] = linesDS.flatMap(_.split(","))


    //2.转换成kv格式
    val kvDS: DataStream[(String, Int)] = wordsDS.map((_, 1))


    //3.按照单词分组
    val keyDS: KeyedStream[(String, Int), String] = kvDS.keyBy(_._1)

    //4.统计单词数量，指定对哪一列求和，指定下标，下标从0开始
    val countDS: DataStream[(String, Int)] = keyDS.sum(1)


    //打印
    countDS.print()

    /**
     *启动
     */


    env.execute()
  }

}

